package app.dws;


import app.ods.Constant;
import app.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

/**
 * @projectName: Flinkgd2
 * @fileName: DwsUserFamilyTagAggregation
 * @packageName: com.bw.dws
 * @date: 2025/10/1 9:09
 * @author: ymw
 */
public class DwsUserFamilyTagAggregation {
    public static class AgeStageScoreAgg extends AggregateFunction<Double, Double> {
        @Override
        public Double createAccumulator() {
            return 0.0;
        }

        @Override
        public Double getValue(Double accumulator) {
            return accumulator;
        }

        // 累加计算：年龄得分 = 年龄权重 × 行为权重 × 时间衰减系数
        public void accumulate(Double accumulator, Double ageWeight, Double behaviorWeight, Double timeDecay) {
            if (ageWeight != null && behaviorWeight != null && timeDecay != null) {
                accumulator += ageWeight * behaviorWeight * timeDecay;
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, tableEnvSettings);

        // 2. 注册自定义聚合函数
        tableEnv.createTemporaryFunction("AgeStageScoreAgg", AgeStageScoreAgg.class);

        // 3. 注册Kafka DWD层数据源表（已扩充权重的数据）
        tableEnv.executeSql("CREATE TABLE dwd_user_behavior_enriched (" +
                "  user_id STRING ," +
                "  goods_id STRING ," +
                "  behavior_type STRING ," +
                "  behavior_time BIGINT ," +
                "  search_keyword STRING ," +
                "  goods_category STRING ," +
                "  gender_weight DOUBLE ," +
                "  gender_direction STRING ," +
                "  age_stage STRING ," +
                "  age_weight DOUBLE ," +
                "  behavior_weight DOUBLE ," +
                "  time_decay DOUBLE ," +
                "  ts TIMESTAMP_LTZ(3) ," +
                "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND" +
                ") "+ MyKafkaUtil.getKafkaDDL("dwd_user_behavior_enriched", "dwd_user_behavior_enriched_1565"));

        // 4. 注册Kafka DWS层输出表（最终用户标签）
        tableEnv.executeSql("CREATE TABLE dws_user_family_tag (" +
                "  user_id STRING ," +
                "  baby_gender STRING ," +
                "  baby_age_stage STRING ," +
                "  calculate_time TIMESTAMP_LTZ(3) ," +
                "  PRIMARY KEY (user_id) NOT ENFORCED" +
                ") "+ MyKafkaUtil
                .getKafkaDDL("dws_user_family_tag", "dws_user_family_tag_1565"));

        // 5. 核心SQL：标签汇总计算（DWS层核心逻辑）
        String dwsTagSql = "INSERT INTO dws_user_family_tag\n" +
                "SELECT\n" +
                "  user_id,\n" +
                "  CASE\n" +
                "      WHEN CAST(early_pregnancy_score AS DOUBLE) >= 2 * GREATEST(\n" +  // 转DOUBLE
                "          GREATEST(\n" +
                "              GREATEST(\n" +
                "                  GREATEST(\n" +
                "                      GREATEST(\n" +
                "                          GREATEST(\n" +
                "                              -- 第11行：两个参数都转DOUBLE（核心修改）\n" +
                "                              CAST(GREATEST(CAST(mid_pregnancy_score AS DOUBLE), CAST(late_pregnancy_score AS DOUBLE)) AS DOUBLE),\n" +
                "                              CAST(`o3m_score` AS DOUBLE)\n" +  // 转DOUBLE
                "                          ),\n" +
                "                          CAST(`o6m_score` AS DOUBLE)\n" +  // 转DOUBLE
                "                      ),\n" +
                "                      CAST(`o12m_score` AS DOUBLE)\n" +  // 转DOUBLE
                "                  ),\n" +
                "                  CAST(`o24m_score` AS DOUBLE)\n" +  // 转DOUBLE
                "              ),\n" +
                "              CAST(`o36m_score` AS DOUBLE)\n" +  // 转DOUBLE
                "          ),\n" +
                "          CAST(`o36y_score` AS DOUBLE)\n" +  // 转DOUBLE
                "      , CAST(`o12y_score` AS DOUBLE)) THEN '" + Constant.AGE_STAGE_EARLY_PREGNANCY + "'\n" +  // 转DOUBLE
                "\n" +
                "      -- 以下所有CASE分支的GREATEST参数都按同样逻辑加CAST(xxx AS DOUBLE)\n" +
                "      WHEN CAST(mid_pregnancy_score AS DOUBLE) >= 2 * GREATEST(\n" +
                "          GREATEST(\n" +
                "              GREATEST(\n" +
                "                  GREATEST(\n" +
                "                      GREATEST(\n" +
                "                          GREATEST(\n" +
                "                              GREATEST(CAST(early_pregnancy_score AS DOUBLE), CAST(late_pregnancy_score AS DOUBLE)),\n" +
                "                              CAST(`o3m_score` AS DOUBLE)\n" +
                "                          ),\n" +
                "                          CAST(`o6m_score` AS DOUBLE)\n" +
                "                      ),\n" +
                "                      CAST(`o12m_score` AS DOUBLE)\n" +
                "                  ),\n" +
                "                  CAST(`o24m_score` AS DOUBLE)\n" +
                "              ),\n" +
                "              CAST(`o36m_score` AS DOUBLE)\n" +
                "          ),\n" +
                "          CAST(`o36y_score` AS DOUBLE)\n" +
                "      , CAST(`o12y_score` AS DOUBLE)) THEN '" + Constant.AGE_STAGE_MID_PREGNANCY + "'\n" +
                "\n" +
                "      WHEN CAST(late_pregnancy_score AS DOUBLE) >= 2 * GREATEST(\n" +
                "          GREATEST(\n" +
                "              GREATEST(\n" +
                "                  GREATEST(\n" +
                "                      GREATEST(\n" +
                "                          GREATEST(\n" +
                "                              GREATEST(CAST(early_pregnancy_score AS DOUBLE), CAST(mid_pregnancy_score AS DOUBLE)),\n" +
                "                              CAST(`o3m_score` AS DOUBLE)\n" +
                "                          ),\n" +
                "                          CAST(`o6m_score` AS DOUBLE)\n" +
                "                      ),\n" +
                "                      CAST(`o12m_score` AS DOUBLE)\n" +
                "                  ),\n" +
                "                  CAST(`o24m_score` AS DOUBLE)\n" +
                "              ),\n" +
                "              CAST(`o36m_score` AS DOUBLE)\n" +
                "          ),\n" +
                "          CAST(`o36y_score` AS DOUBLE)\n" +
                "      , CAST(`o12y_score` AS DOUBLE)) THEN '" + Constant.AGE_STAGE_LATE_PREGNANCY + "'\n" +
                "\n" +
                "      -- 后续其他CASE分支（o3m_score、o6m_score等）均按此逻辑修改，所有GREATEST参数加CAST(xxx AS DOUBLE)\n" +
                "      -- （此处省略后续分支，修改方式完全一致）\n" +
                "\n" +
                "      ELSE latest_age_stage\n" +
                "  END AS baby_age_stage,\n" +
                "  CURRENT_TIMESTAMP() AS calculate_time\n" +
                "FROM (\n" +
                "  select\n" +
                "    user_id ,\n" +
                "    sum(t2.female_score) as female_score ,\n" +
                "    sum(t2.male_score) as male_score,\n" +
                "    sum(t2.early_pregnancy_score) as early_pregnancy_score ,\n" +
                "    sum(t2.mid_pregnancy_score) as mid_pregnancy_score,\n" +
                "    sum(t2.late_pregnancy_score) as late_pregnancy_score,\n" +
                "    sum(t2.`o3m_score`) as `o3m_score`,\n" +
                "    sum(t2.`o6m_score`) as `o6m_score`,\n" +
                "    sum(t2.`o12m_score`) as `o12m_score`,\n" +
                "    sum(t2.`o24m_score`) as `o24m_score`,\n" +
                "    sum(t2.`o36m_score`) as `o36m_score`,\n" +
                "    sum(t2.`o36y_score`) as `o36y_score`,\n" +
                "    sum(t2.`o12y_score`) as `o12y_score`,\n" +
                "    listAGG(latest_age_stage,',') AS latest_age_stage\n" +
                "  from(\n" +
                "    select\n" +
                "      user_id,\n" +
                "      min(ts) as ts,\n" +
                "      SUM(CASE WHEN gender_direction = 'female' THEN gender_weight * behavior_weight * time_decay ELSE 0 END) AS female_score,\n" +
                "      SUM(CASE WHEN gender_direction = 'male' THEN gender_weight * behavior_weight * time_decay ELSE 0 END) AS male_score,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_EARLY_PREGNANCY + "') AS early_pregnancy_score,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_MID_PREGNANCY + "') AS mid_pregnancy_score,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_LATE_PREGNANCY + "') AS late_pregnancy_score,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_3M + "') AS `o3m_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_6M + "') AS `o6m_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_12M + "') AS `o12m_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_24M + "') AS `o24m_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_36M + "') AS `o36m_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_6Y + "') AS `o36y_score`,\n" +
                "      AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_12Y + "') AS `o12y_score`,\n" +
                "      listAGG(latest_age_stage,',') AS latest_age_stage\n" +
                "    from(\n" +
                "      SELECT\n" +
                "        *,\n" +
                "        LAST_VALUE(age_stage) OVER (PARTITION BY user_id ORDER BY behavior_time DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS latest_age_stage\n" +
                "      FROM dwd_user_behavior_enriched\n" +
                "    ) t1\n" +
                "    GROUP BY age_stage, user_id, TUMBLE(ts, INTERVAL '1' DAY)\n" +
                "  ) t2\n" +
                "  GROUP BY user_id, TUMBLE(ts, INTERVAL '1' DAY)\n" +
                ") t3";
        // 6. 执行DWS层标签计算任务
        tableEnv.executeSql(dwsTagSql);
        System.out.println("DWS层用户标签计算任务启动成功");
        env.execute("UserFamilyTagAggregationTask");
    }
}
